package cdrprocessing.bolt.util;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import cdrprocessing.common.CallPeriod;

public class ThrottlingBolt extends BaseRichBolt {

    private List<CdrTuple> tuples;

    private int rate;

    private OutputCollector output;

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("CdrStream", new Fields("customer", "area", "duration", "period", "terminateReason"));
    }

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        rate = Integer.valueOf((String) stormConf.get("outputRate"));
        tuples = new ArrayList<CdrTuple>(rate);
        output = collector;
    }

    public void execute(Tuple input) {
        storeTuple(input);

        if (tuples.size() == rate) {
            final Iterator<CdrTuple> it = tuples.iterator();
            while (it.hasNext()) {
                final CdrTuple tuple = it.next();
                output.emit("CdrStream", new Values(tuple.getCustomer(), tuple.getArea(), tuple.getDuration(), tuple.getPeriod(), tuple.getTerminateReason()));
                it.remove();
            }
        }
    }

    private void storeTuple(Tuple input) {
        final CdrTuple tuple = new CdrTuple();
        tuple.setCustomer(input.getLongByField("customer"));
        tuple.setArea(input.getIntegerByField("area"));
        tuple.setDuration(input.getDoubleByField("duration"));
        tuple.setPeriod((CallPeriod) input.getValueByField("period"));
        tuple.setTerminateReason(input.getIntegerByField("terminateReason"));

        tuples.add(tuple);
    }

    private final class CdrTuple {

        private long customer;

        private int area;

        private double duration;

        private CallPeriod period;

        private int terminateReason;

        public long getCustomer() {
            return customer;
        }

        public void setCustomer(long customer) {
            this.customer = customer;
        }

        public int getArea() {
            return area;
        }

        public void setArea(int area) {
            this.area = area;
        }

        public double getDuration() {
            return duration;
        }

        public void setDuration(double duration) {
            this.duration = duration;
        }

        public CallPeriod getPeriod() {
            return period;
        }

        public void setPeriod(CallPeriod period) {
            this.period = period;
        }

        public int getTerminateReason() {
            return terminateReason;
        }

        public void setTerminateReason(int terminateReason) {
            this.terminateReason = terminateReason;
        }

    }
}
